package com.wmx.rocketmq.helloWorld;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 异步发送消息 - 异步传输通常用于响应时间敏感的业务场景
 *
 * @author wangMaoXiong
 * @version 1.0
 * @date 2020/11/2 19:43
 */
public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        //声明并初始化一个 producer 生产者，同时指定 Producer Group 生产者组名称
        DefaultMQProducer producer = new DefaultMQProducer("rocketMQ-helloWorld-asyncProducerGroup");

        //1、指定 Producer 连接的 nameServer 服务器所在地址以及端口
        //2、如果是分布式部署的多个，则用分号隔开，如：setNamesrvAddr("172.16.235.77:9876;172.16.235.78:9876");
        producer.setNamesrvAddr("192.168.116.128:9876");

        //Producer 对象在使用之前必须要调用 start 进行启动初始化，初始化一次即可，切忌不可每次发送消息时，都调用 start 方法
        producer.start();

        //在声明异步模式下发送失败之前内部执行的最大重试次数。
        //这可能会导致消息重复，这由应用程序开发人员来解决，默认为 2
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 100;
        //因为是异步发送，所以借助 CountDownLatch 倒计数锁存器来防止 main主线程 提前结束退出
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                //主题、标签等等全部自定义即可
                Message msg = new Message("TopicTest", "TagA", "OrderID188",
                        ("我是异步消息" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                //异步发送消息，分别对发送成功和失败进行处理
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf("%-3d Success %s %n", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        //应用退出时，调用 shutdown 关闭网络连接，清理资源，从 RocketMQ 服务器上注销自己
        //建议应用在 JBOSS、Tomcat 等容器的退出钩子里调用 shutdown 方法
        producer.shutdown();
        System.out.printf("Producer shutdown.%n");
    }
}
